今天又是美好的一天，讓我們先來看看為啥昨天 Agent 跑的 pytest 一直有好幾個 skip。

(e-fdhs) C:\Users\zalic\desktop\e-fdhs\backend>python -m pytest
=== test session starts ===
platform win32 -- Python 3.12.14, pytest-8.4.2, pluggy-1.6.0
rootdir: C:\Users\zalic\desktop\e-fdhs\backend
plugins: anyio-4.15.1, asyncio-0.26.0
asyncio: mode=Mode.STRICT, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collected 10 items

tests\test_auth_helpers.py ..               [ 20%]
tests\test_auth_routes.py s                 [ 30%]
tests\test_authorization_services.py sssss  [ 80%]
tests\test_route_loader.py ..               [100%]

=== 4 passed, 6 skipped in 1.14s ===

對阿，這六個 skipped 是怎樣，來看看原因。

(e-fdhs) C:\Users\zalic\desktop\e-fdhs\backend>python -m pytest -rs
=== test session starts ===
platform win32 -- Python 3.12.14, pytest-8.4.2, pluggy-1.6.0
rootdir: C:\Users\zalic\desktop\e-fdhs\backend
plugins: anyio-4.15.1, asyncio-0.26.0
asyncio: mode=Mode.STRICT, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collected 10 items

tests\test_auth_helpers.py ..     [ 20%]
tests\test_auth_routes.py s       [ 30%]
tests\test_authorization_services.py sssss    [ 80%]
tests\test_route_loader.py ..     [100%]

=== short test summary info ===
SKIPPED [1] tests\test_auth_routes.py:57: PostgreSQL integration tests require TEST_DB_HOST, TEST_DB_PORT, TEST_DB_NAME, TEST_DB_USER, and TEST_DB_PASSWORD
SKIPPED [1] tests\test_authorization_services.py:51: PostgreSQL integration tests require TEST_DB_HOST, TEST_DB_PORT, TEST_DB_NAME, TEST_DB_USER, and TEST_DB_PASSWORD
SKIPPED [1] tests\test_authorization_services.py:90: PostgreSQL integration tests require TEST_DB_HOST, TEST_DB_PORT, TEST_DB_NAME, TEST_DB_USER, and TEST_DB_PASSWORD
SKIPPED [1] tests\test_authorization_services.py:126: PostgreSQL integration tests require TEST_DB_HOST, TEST_DB_PORT, TEST_DB_NAME, TEST_DB_USER, and TEST_DB_PASSWORD
SKIPPED [1] tests\test_authorization_services.py:153: PostgreSQL integration tests require TEST_DB_HOST, TEST_DB_PORT, TEST_DB_NAME, TEST_DB_USER, and TEST_DB_PASSWORD
SKIPPED [1] tests\test_authorization_services.py:182: PostgreSQL integration tests require TEST_DB_HOST, TEST_DB_PORT, TEST_DB_NAME, TEST_DB_USER, and TEST_DB_PASSWORD
=== 4 passed, 6 skipped in 1.08s ===

嘶...…環境變數？

但我 .env 有寫啊。

來看看 requirements。

fastapi>=0.115,<1.0
uvicorn[standard]>=0.30,<1.0
SQLAlchemy>=2.0,<2.1
asyncpg>=0.29,<1.0
alembic>=1.13,<2.0
pytest>=8.0,<9.0
pytest-asyncio>=0.23,<1.0
argon2-cffi>=23.1,<26.0
httpx>=0.27,<1.0

嘶...…給我幾秒，我要破案了。

……

不對阿，dotenv 呢？

來去看看 database.py。

"""SQLAlchemy asynchronous PostgreSQL configuration."""

from __future__ import annotations

import os
from collections.abc import AsyncGenerator

from sqlalchemy.engine import URL
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

我去，老哥，你讓程式通靈呢？
不用 dotenv，就想讓它自己讀 .env？那很厲害了。

我補一下。

from __future__ import annotations

import os
import dotenv
from collections.abc import AsyncGenerator

from sqlalchemy.engine import URL
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

dotenv.load_dotenv()

這下應該可以了。

(e-fdhs) C:\Users\zalic\desktop\e-fdhs\backend>python -m pytest -rs
=== test session starts ===
platform win32 -- Python 3.12.14, pytest-8.4.2, pluggy-1.6.0
rootdir: C:\Users\zalic\desktop\e-fdhs\backend
plugins: anyio-4.15.1, asyncio-0.26.0
asyncio: mode=Mode.STRICT, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collected 10 items

tests\test_auth_helpers.py ..        [ 20%]
tests\test_auth_routes.py s          [ 30%]
tests\test_authorization_services.py sssss          [ 80%]
tests\test_route_loader.py ..        [100%]

=== short test summary info ===
SKIPPED [1] tests\test_auth_routes.py:57: PostgreSQL integration tests require TEST_DB_HOST, TEST_DB_PORT, TEST_DB_NAME, TEST_DB_USER, and TEST_DB_PASSWORD
SKIPPED [1] tests\test_authorization_services.py:51: PostgreSQL integration tests require TEST_DB_HOST, TEST_DB_PORT, TEST_DB_NAME, TEST_DB_USER, and TEST_DB_PASSWORD
SKIPPED [1] tests\test_authorization_services.py:90: PostgreSQL integration tests require TEST_DB_HOST, TEST_DB_PORT, TEST_DB_NAME, TEST_DB_USER, and TEST_DB_PASSWORD
SKIPPED [1] tests\test_authorization_services.py:126: PostgreSQL integration tests require TEST_DB_HOST, TEST_DB_PORT, TEST_DB_NAME, TEST_DB_USER, and TEST_DB_PASSWORD
SKIPPED [1] tests\test_authorization_services.py:153: PostgreSQL integration tests require TEST_DB_HOST, TEST_DB_PORT, TEST_DB_NAME, TEST_DB_USER, and TEST_DB_PASSWORD
SKIPPED [1] tests\test_authorization_services.py:182: PostgreSQL integration tests require TEST_DB_HOST, TEST_DB_PORT, TEST_DB_NAME, TEST_DB_USER, and TEST_DB_PASSWORD
=== 4 passed, 6 skipped in 1.01s ===

？？？？？？

6。

你不是說破案了？

不是，等一下。

.env 有寫。

dotenv 也裝了。

load_dotenv() 也確實跑了。

那這六個到底為什麼還在 skip？

再仔細看看剛才的訊息。

PostgreSQL integration tests require
TEST_DB_HOST, TEST_DB_PORT, TEST_DB_NAME,
TEST_DB_USER, and TEST_DB_PASSWORD

但這些東西我明明也有寫在 .env 裡。

TEST_DB_HOST=localhost
TEST_DB_PORT=5432
TEST_DB_NAME=e-fdhs-test
TEST_DB_USER=e-fdhs-test
TEST_DB_PASSWORD=***

那有沒有一種可能……

你的程式又在通靈？

不可能。

這次我真的 load_dotenv() 了。

既然環境變數有寫、dotenv 也有載入，那就只好看看這些測試到底是怎麼判斷資料庫有沒有設定了。

翻一下 conftest.py。

_DATABASE_KEYS = ("DB_HOST", "DB_PORT", "DB_NAME", "DB_USER", "DB_PASSWORD")
_MISSING_TEST_KEYS = [key for key in _DATABASE_KEYS if not os.getenv(f"TEST_{key}")]

POSTGRES_TEST_CONFIGURED = not _MISSING_TEST_KEYS
os.environ["POSTGRES_TEST_CONFIGURED"] = "1" if POSTGRES_TEST_CONFIGURED else "0"

……

喔。

好像真的又破案了。

不是沒讀，是讀太早了

我們剛才把 load_dotenv() 寫在哪？

database.py。

但 pytest 在跑測試之前，會先載入 conftest.py。

而 conftest.py 一被載入，就立刻執行了這段：

_MISSING_TEST_KEYS = [
    key for key in _DATABASE_KEYS
    if not os.getenv(f"TEST_{key}")
]

問題來了。

這個時候 database.py 根本還沒幫我們載入 .env。

所以實際上的順序比較像這樣：

pytest
  ↓
載入 conftest.py
  ↓
os.getenv("TEST_DB_HOST")
  ↓
None
  ↓
「測試資料庫沒設定！」
  ↓
POSTGRES_TEST_CONFIGURED = False
  ↓
後面才 import database.py
  ↓
load_dotenv()
  ↓
「欸，我現在有環境變數了」

那不是有了嗎？

有啊。

但是太晚了。

前面都已經判你沒設定測試資料庫了，現在才把 .env 載進來有什麼用。

load_dotenv() 又不是神秘妙妙工具。

事後來看，剛才把 load_dotenv() 補進 database.py，只解決了應用程式載入環境變數的問題；但 conftest.py 早在這之前，就已經把測試資料庫「有沒有設定」這件事判斷完了。

所以你剛剛不是沒讓程式讀 .env。

是考卷交出去之後才想起來要寫名字？

……

超好笑。

總之，既然 conftest.py 自己就需要讀取 TEST_DB_*，那它就必須在讀取之前先確保 .env 已經被載入。

所以再補一次：

from __future__ import annotations

import os
import dotenv

import pytest

dotenv.load_dotenv()

這次順序終於正常了：

載入 conftest.py
        ↓
    load_dotenv()
        ↓
載入 TEST_DB_*
        ↓
檢查 TEST_DB_*
        ↓
設定測試資料庫
        ↓
開始跑 Integration Test

好。

這次總該破案了吧？

你上次也是這麼說的。

……

跑了再說。

(e-fdhs) C:\Users\zalic\desktop\e-fdhs\backend>python -m pytest
=== test session starts ===
platform win32 -- Python 3.12.14, pytest-8.4.2, pluggy-1.6.0
rootdir: C:\Users\zalic\desktop\e-fdhs\backend
plugins: anyio-4.15.1, asyncio-0.26.0
asyncio: mode=Mode.STRICT, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collected 10 items

tests\test_auth_helpers.py ..                       [ 20%]
tests\test_auth_routes.py F                         [ 30%]
tests\test_authorization_services.py .E.EF          [ 80%]
tests\test_route_loader.py ..                       [100%]

=== ERRORS ===
......中間省略無限多字
======= 2 failed, 6 passed, 2 warnings, 2 errors in 3.50s ========
C:\Users\zalic\miniconda3\envs\e-fdhs\Lib\site-packages\_pytest\unraisableexception.py:33: RuntimeWarning: coroutine 'Connection._cancel' was never awaited
  gc.collect()
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

我有點想念剛剛的六個 skipped 了。

來問問 GPT 好了，這麼多 error 我有點看不過來。

把整坨錯誤丟給 GPT 後，它幫我把目前的問題大概分成了三類。

1. RuntimeError: Event loop is closed
2. sqlalchemy.exc.MissingGreenlet
3. Session 時間差了 184 μs

嗯。

至少從幾百行變成三行了。

等等。

第一個是不是有點眼熟？

哪個？

Event loop is closed。

……

等等。

RuntimeError: Event loop is closed

……幹。

怎麼又是 Event Loop。

前幾天才花了一整篇研究怎麼不要謀殺 Event Loop，今天它就自己躺在我的測試裡了。

有沒有一種可能，它不是自己躺進去的？

你先閉嘴。

Event Loop 怎麼又死了？

讓我們來看看這兇案現場。

把那幾百行 Traceback 往上翻，可以找到這一段：

@pytest_asyncio.fixture(autouse=True)
async def empty_database() -> None:
    async with engine.begin() as connection:
        ...

然後它在 engine.begin() 裡一路炸下去，最後留下：

RuntimeError: Event loop is closed

奇怪了。

這裡也沒有 time.sleep()，沒有什麼同步函式卡住 Event Loop，更沒有哪個天才手動寫：

loop.close()

那它到底怎麼死的？

再回去看看剛才 pytest 開頭其實已經給過我們一條線索。

asyncio: mode=Mode.STRICT,
asyncio_default_fixture_loop_scope=None,
asyncio_default_test_loop_scope=function

最後那個：

asyncio_default_test_loop_scope=function

意思是這些 async test 預設不是從頭到尾共用同一個 Event Loop。

每個 test 都可以有自己的 Event Loop。

大概像這樣：

test A
└── Event Loop A
    └── 測試結束
        └── Event Loop A 關閉

test B
└── Event Loop B

看起來很合理。

每場考試發一張新的考卷，用完收走，下一個人再拿一張新的。

問題是，我們的資料庫連線不一定跟著換。

前面建立 SQLAlchemy Engine 的時候，我們是直接在 module 裡建立一個共用的 engine。

而 AsyncEngine 背後還有 Connection Pool。

簡單來說，資料庫連線不是每次用完就一定直接丟進垃圾桶；可以先留在 Pool 裡，之後需要的時候再拿出來用。

於是就有機會發生一件非常尷尬的事情：

Test A
│
├── Event Loop A
│   └── 建立資料庫 Connection
│
├── Connection 放回 Pool
│
└── 測試結束
    └── Event Loop A 關閉

Test B
│
├── Event Loop B
│
└── 從 Pool 拿出剛才的 Connection
    │
    └── 「哈囉？」
         「你原本的 Event Loop 呢？」
         「死了。」

……

所以你前幾天才叫我不要謀殺 Event Loop。

結果你今天直接拿亡者的遺物給下一個 Event Loop 用？

這個形容聽起來有點恐怖，但差不多是這個意思。

問題並不是 async with engine.begin() 把 Event Loop 關掉了。

而是測試之間換了 Event Loop，但同一個 AsyncEngine 的 Connection Pool 還可能留著上一個 Event Loop 用過的 Connection。

下一個 test 如果又重用了 Pool 裡、上一個 Event Loop 使用過的 asyncpg Connection，就可能在操作或清理它時碰上一個已經關閉的 Event Loop。

然後：

RuntimeError: Event loop is closed

好。

至少這次不是我在 async def 裡面塞 time.sleep() 了。

值得驕傲嗎？

不值得。

但至少我們現在知道屍體怎麼來的了。

總之，我不是很想自己修，所以我們一樣把問題交給 agent 吧~

那就別把遺物留給下一個 Event Loop

既然問題是上一個 Event Loop 用過的 Connection 還留著，那解法其實很直覺。

用完就清掉。

SQLAlchemy 的 AsyncEngine 有一個 dispose()，可以把目前 Pool 裡留著的 Connection 清掉。

所以 Agent 在 conftest.py 加了一個 fixture：

@pytest_asyncio.fixture(autouse=True)
async def dispose_async_engine_after_test() -> None:
    yield
    await engine.dispose()

等等，yield 又是什麼？

這裡先把它想成：

測試前
  ↓
yield
  ↓
開始跑 Test
  ↓
Test 結束
  ↓
回來繼續執行 yield 下面的程式

所以這段的意思其實就是：

每個 Test 跑完之後，執行一次 engine.dispose()。

這樣上一個 Event Loop 用過的 Connection 就不會繼續留在 Pool 裡，等著下一個 Test 過來撿。

所以你前幾天叫我不要謀殺 Event Loop。
今天則是叫我記得清遺物？

……

可以不要把非同步寫得跟殯葬業一樣嗎？

總之，第一隻處理完。

接下來第二隻。

幾百行錯誤，只是因為晚拿了一個 ID？

sqlalchemy.exc.MissingGreenlet

Missing……什麼？

Greenlet。

那是什麼？

……

其實這次我們不用知道。

先看它在哪裡爆炸：

await database_session.rollback()

async with database_session.begin():
    await AccountService(database_session).update_password_without_commit(
        principal.account.id,
        password_hash=await hash_password(payload.new_password),
    )

看起來沒什麼問題。

但注意一下順序：

rollback()
    ↓
principal.account.id

principal.account 是 SQLAlchemy 管理的 ORM Object。

而在 rollback() 之後，這個物件裡原本讀到的資料，可能會被 SQLAlchemy 視為需要重新確認。

於是當我寫：

principal.account.id

我以為自己只是在拿一個普通的 Python 屬性。

SQLAlchemy 卻有可能想：

等等，這份資料我要重新去資料庫拿。

問題來了。

查資料庫是 I/O。

我們現在用的又是 Async SQLAlchemy。

那理論上是不是應該要有：

await ...

但我現在寫的是：

principal.account.id

這只是一個普通的屬性存取，哪來的 await？

於是 SQLAlchemy 想做非同步 I/O，卻發現現在不是能這樣做的地方。

然後：

sqlalchemy.exc.MissingGreenlet

所以你 rollback() 完，又跑回去跟 SQLAlchemy 要資料？

……

好像是這樣。

那你早點拿不就好了？

……

好像也是這樣。

所以最後的修法是：

account_id = principal.account.id

await database_session.rollback()

async with database_session.begin():
    await AccountService(database_session).update_password_without_commit(
        account_id,
        password_hash=await hash_password(payload.new_password),
    )
    await SessionService(database_session).revoke_all_for_account_without_commit(
        account_id
    )

在 rollback() 之前，先把等等還要用的 account_id 存成普通的 Python 值。

後面就不用再碰那個 ORM Object。

就這樣？

就這樣。

幾百行 Traceback。

只要搬一行就能解決。

等等，不是還有第三隻嗎？

哪隻？

184 μs。

……

那隻太小了，放生。

明天見。